package com.wsx.gradeanaly;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class TestMapper extends Mapper<LongWritable, Text, Text, Student> {

	private float max = Float.MIN_VALUE;
	private float min = Float.MAX_VALUE;
	private Student maxstu = new Student();
	private Student minstu = new Student();
	private Student ordistu;
	
	private String[] class_file_path = {"hdfs://localhost:9000/user/root/input/one", "hdfs://localhost:9000/user/root/input/two"};
	private long[] num = new long[2];
	private double[] sum = new double[2];
    @Override
	protected void map(LongWritable key, Text value,
			Mapper<LongWritable, Text, Text, Student>.Context context)
			throws IOException, InterruptedException {
    	String line = value.toString();
    	if (line == null || line.equals("")) {  
            return;  
        }
    	String[] strs = line.split(" ");
    	float temp = Float.parseFloat(strs[1]);
    	
    	InputSplit  inputSplit = context.getInputSplit();
		String fileName = ((FileSplit) inputSplit).getPath() .toString();
		
		for(int i=0; i<class_file_path.length; i++){
			if((class_file_path[i]).equals(fileName)){
				sum[i]+=temp;
				num[i]+=1;
			}
		}
    	
    	ordistu = new Student(strs[0], temp, GetRank(temp));
    	context.write(new Text("common"), ordistu);
    	
    	if(temp > max){
    		max = temp;
    		maxstu.setRank(GetRank(max));
    		maxstu.setScore(max);
    		maxstu.setName(strs[0]);
    	}else if(temp < min){
    		min = temp;
    		minstu.setRank(GetRank(min));
    		minstu.setScore(min);
    		minstu.setName(strs[0]);
    	}
	}
    
	@Override
	protected void cleanup(Mapper<LongWritable, Text, Text, Student>.Context context)
			throws IOException, InterruptedException {
        context.write(new Text("max"), maxstu);
        context.write(new Text("min"), minstu);
        
        InputSplit  inputSplit = context.getInputSplit();
		String fileName = ((FileSplit) inputSplit).getPath() .toString();
		for(int i=0; i<class_file_path.length; i++){
			if((class_file_path[i]).equals(fileName)){
				context.write(new Text("aver"), new Student(String.valueOf(sum[i])+";"+String.valueOf(num[i]), 0, 'F'));
			}
		}
	}
	
	public char GetRank(float temp){
		int cla = (int)(temp / 10);
		if(cla == 10)
			return 'A';
		if(cla == 9)
			return 'A';
		if(cla == 8)
			return 'B';
		if(cla == 7)
			return 'C';
		if(cla == 6)
			return 'D';
		return 'E';
	}
	
}
